package com.googlecode.xcarve.router;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.transport.Client;

public final class ConsistentHashRouter extends AbstractRouter{
	
	
	private final ConcurrentMap<String, ConsistentHashSelector> selectors = new ConcurrentHashMap<String, ConsistentHashSelector>();
	static  final Logger logger = LoggerFactory.getLogger(ConsistentHashRouter.class);
	
	/* (non-Javadoc)
	 * @see com.dove.transport.router.AbstractRouter#doSelect(java.util.List, java.lang.String)
	 */
	@Override
	Client doSelect(List<? extends Client> list,String service, String hashKey) {
		ConsistentHashSelector select = selectors.get(service);
		if(select == null){
			selectors.put(service, new ConsistentHashSelector(list));
			select = selectors.get(service);
		}
		return select.get(hashKey);
	}
 
	private final class ConsistentHashSelector{
		static final int NUM_REPS = 160;
		private final TreeMap<Long, Client> virtualInvokers;
		ConsistentHashSelector(final List<? extends Client> list){
			virtualInvokers = new TreeMap<Long, Client>();
			buildMap(list);
		}
		
		void buildMap(final List<? extends Client> list){
			for(Client client : list){
				for (int i = 0; i < NUM_REPS / 4; i++) {
					byte[] digest = md5(client.getConnection().remoteAddress().toString() + "-" + i);
					for (int h = 0; h < 4; h++) {
						long k = hash(digest, h);
						virtualInvokers.put(k, client);
					}
				}
			}
		}
		
		public Client get(String serviceKey){
			long hash = hash(md5(serviceKey),0);
			if (!virtualInvokers.containsKey(hash)) {
				// Java 1.6 adds a ceilingKey method, but for compatible with jdk5,
				// So use tailMap method to do this.
				SortedMap<Long, Client> tailMap = virtualInvokers.tailMap(hash);
				if (tailMap.isEmpty()) {
					hash = virtualInvokers.firstKey();
				} else {
					hash = tailMap.firstKey();
				}
			}
//			Client client = virtualInvokers.get(hash);
//			int tries = 0;
//			int max =4;
//			while((client == null || client.isClosed()) && tries++ < max){
//				
//			}
			return virtualInvokers.get(hash);
		}
		
		private long hash(byte[] digest, int number) {
	            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
	                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
	                    | ((long) (digest[1 + number * 4] & 0xFF) << 8) 
	                    | (digest[0 + number * 4] & 0xFF)) 
	                    & 0xFFFFFFFFL;
	    }
	
		private byte[] md5(String value) {
			MessageDigest md5;
			try {
				md5 = MessageDigest.getInstance("MD5");
			} catch (NoSuchAlgorithmException e) {
				throw new IllegalStateException(e.getMessage(), e);
			}
			md5.reset();
			byte[] bytes = null;
			try {
				bytes = value.getBytes("UTF-8");
			} catch (UnsupportedEncodingException e) {
				throw new IllegalStateException(e.getMessage(), e);
			}
			md5.update(bytes);
			return md5.digest();
		}
	}

	public void update(final List<? extends Client> list,String service){
		selectors.put(service, new ConsistentHashSelector(list));
		logger.info("xcarve has update ConsistentHashSelector successfull. service={}",service);
	}
	
}
